[observability][export-api] Write submission job events #47468

nikitavemuri · 2024-09-03T21:52:33Z

Why are these changes needed?

Add ExportEventLoggerAdapter which will be used to write export events to file from python files. Only a single ExportEventLoggerAdapter instance will exist per source type, so callers can create or get this instance using get_export_event_logger which is thread safe.
Write Submission Job export events to file from JobInfoStorageClient.put_info which is called to update the JobInfo data in the internal KV store.

Related issue number

Checks

I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
I've run scripts/format.sh to lint the changes in this PR.
I've included any doc changes needed for https://docs.ray.io/en/master/.
- I've added any new APIs to the API Reference. For example, if I added a
  method in Tune, I've added it in doc/source/tune/api/ under the
  corresponding .rst file.
I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
Testing Strategy
- Unit tests
- Release tests
- This PR is not tested :(

Signed-off-by: Nikita Vemuri <[email protected]>

…ita/write_job_submission_events

Signed-off-by: Nikita Vemuri <[email protected]>

alanwguo · 2024-09-05T23:00:45Z

python/ray/_private/event/export_event_logger.py

+ExportEventDataType = Union[ExportSubmissionJobEventData]
+
+
+def get_event_id():


nit

Suggested change

def get_event_id():

def generate_event_id():

alanwguo · 2024-09-05T23:02:14Z

python/ray/_private/event/export_event_logger.py

+        # Force flush so that we won't lose events
+        self.logger.handlers[0].flush()


is this a pattern we follow in the other event exporters as well?

I think batching flushing is reasonable if we get significant performance improvements at the loss of some reliability

since it is only used by a job, it is unlikely cause a problem, but I agree it is probably okay to lose some events here... (I don't think we guarantee flush every event after it is finished)

Yeah, I added this behavior for consistency with the EventLoggerAdapter for existing python events

ray/python/ray/_private/event/event_logger.py

Line 105 in d8a85c5

self.logger.handlers[0].flush()

The LogEventReporter for C++ events also flushes after each event by default

ray/src/ray/util/event.h

Line 103 in d8a85c5

bool force_flush = true,

The export events from the python layer are unlikely to be created with very high volume (unlike tasks), so we can probably just keep this flush logic for now for consistency and optimize later if needed.

alanwguo · 2024-09-05T23:12:34Z

python/ray/_private/event/export_event_logger.py

+_export_event_logger = {}
+
+
+def get_export_event_logger(source: ExportEvent.SourceType, sink_dir: str):


can you check how we do structured logging in python layer and see if we should follow similar patterns?

Do they also use a singleton pattern like this?

Structured logging fetches various registered loggers for the different source types, but the existing python events use this singleton pattern. We want to use a single logger to avoid any concurrency issues from multiple threads. Using register logger is an option which also is a singleton, but I think this API is cleaner because there is no separate initialization and get functions.

rkooo567 · 2024-09-05T23:44:36Z

python/ray/dashboard/modules/job/tests/test_job_manager.py

+    Test submission job events are correctly generated and written to file
+    as the job goes through various state changes in its lifecycle.
+    """
+    ray_constants.RAY_ENABLE_EXPORT_API_WRITE = True


this screws up the global variable for the rest of the tests. you should probably create a fixture if you want to do this. (or alternatively, just update os.environ)

Updated to set the environment variable in addition to passing the env var when calling ray start, because the constant needs to be set outside ray tasks also when submitting a job.

rkooo567 · 2024-09-05T23:47:10Z

python/ray/_private/event/export_event_logger.py

+    filepath.touch(exist_ok=True)
+    # Configure the logger.
+    handler = logging.handlers.RotatingFileHandler(
+        filepath, maxBytes=(100 * 1e6), backupCount=20  # 100 MB max file size


make it configurable via env var?

rkooo567 · 2024-09-05T23:48:40Z

python/ray/_private/event/export_event_logger.py

+        event = ExportEvent()
+        event.event_id = get_event_id()
+        event.timestamp = int(datetime.now().timestamp())
+        if type(event_data) is ExportSubmissionJobEventData:


Suggested change

if type(event_data) is ExportSubmissionJobEventData:

if isinstance(event_data, ExportSubmissionJobEventData):

rkooo567 · 2024-09-05T23:51:17Z

python/ray/_private/event/export_event_logger.py

+        # Force flush so that we won't lose events
+        self.logger.handlers[0].flush()


since it is only used by a job, it is unlikely cause a problem, but I agree it is probably okay to lose some events here... (I don't think we guarantee flush every event after it is finished)

rkooo567 · 2024-09-05T23:52:29Z

python/ray/dashboard/modules/job/common.py

        self._gcs_aio_client = gcs_aio_client
+        self._export_submission_job_event_logger = None


Add a type!

rkooo567 · 2024-09-05T23:52:52Z

python/ray/dashboard/modules/job/common.py

@@ -189,8 +197,25 @@ class JobInfoStorageClient:
    JOB_DATA_KEY_PREFIX = f"{ray_constants.RAY_INTERNAL_NAMESPACE_PREFIX}job_info_"
    JOB_DATA_KEY = f"{JOB_DATA_KEY_PREFIX}{{job_id}}"

-    def __init__(self, gcs_aio_client: GcsAioClient):
+    def __init__(self, gcs_aio_client: GcsAioClient, log_dir: Optional[str] = None):


let's not allow log_dir = None case?

def __init__(self, gcs_aio_client: GcsAioClient, log_dir: str):

I kept log_dir as an optional variable because the JobInfoStorageClient is called from quite a few places to get job info (no updates to the KV store so no export events are emitted). These callers don’t have a log_dir, so unless we have a global log dir that can be fetched making this variable non optional requires a refactor throughout the jobs code.

let's rename the variable to export_event_log_dir_root? Something where it's clearly meant to control one specific thing.

rkooo567 · 2024-09-05T23:53:57Z

python/ray/dashboard/modules/job/common.py

+        if added_num == 1 or overwrite:
+            # Write export event if data was updated in the KV store
+            try:
+                self._write_submission_job_export_event(job_id, job_info)


QQ: we don't write any event for failure cases?

This isn't expected to fail, but wrapped in a try catch so the remaining job code isn't affected in any way. If there is a failure though, we should log in the normal job log file and not export event log files because the export event files may be ingested by some consumer.

rkooo567 · 2024-09-05T23:54:39Z

python/ray/dashboard/modules/job/common.py

+                job_info.status.name
+            )
+        )
+        if status_value_descriptor is None:


when does it happen?

When a new JobInfo status is added to the python model without updating the proto. We do have a comment to keep these in sync though, so this should be unlikely.

rkooo567 · 2024-09-05T23:54:56Z

python/ray/dashboard/modules/job/tests/test_job_manager.py

@@ -46,6 +48,59 @@
 import psutil


+@pytest.mark.asyncio


can we create a new test suite for all export events? I assume we will need more e2e testing eventually in python layers, and it'd be nice to put them in the same file.

Signed-off-by: Nikita Vemuri <[email protected]>

alanwguo · 2024-09-23T19:39:18Z

python/ray/dashboard/modules/job/common.py

@@ -189,8 +197,25 @@ class JobInfoStorageClient:
    JOB_DATA_KEY_PREFIX = f"{ray_constants.RAY_INTERNAL_NAMESPACE_PREFIX}job_info_"
    JOB_DATA_KEY = f"{JOB_DATA_KEY_PREFIX}{{job_id}}"

-    def __init__(self, gcs_aio_client: GcsAioClient):
+    def __init__(self, gcs_aio_client: GcsAioClient, log_dir: Optional[str] = None):


let's rename the variable to export_event_log_dir_root? Something where it's clearly meant to control one specific thing.

Signed-off-by: Nikita Vemuri <[email protected]>

…47468) Add ExportEventLoggerAdapter which will be used to write export events to file from python files. Only a single ExportEventLoggerAdapter instance will exist per source type, so callers can create or get this instance using get_export_event_logger which is thread safe. Write Submission Job export events to file from JobInfoStorageClient.put_info which is called to update the JobInfo data in the internal KV store. Signed-off-by: ujjawal-khare <[email protected]>

Nikita Vemuri added 6 commits September 3, 2024 00:31

wip

c7c5a5a

Signed-off-by: Nikita Vemuri <[email protected]>

add python logger

e163802

Signed-off-by: Nikita Vemuri <[email protected]>

add schema

3632807

Signed-off-by: Nikita Vemuri <[email protected]>

wip

098b07f

Signed-off-by: Nikita Vemuri <[email protected]>

wip

24fa611

Signed-off-by: Nikita Vemuri <[email protected]>

add test

bdcfff8

Signed-off-by: Nikita Vemuri <[email protected]>

nikitavemuri added the go add ONLY when ready to merge, run all tests label Sep 3, 2024

Nikita Vemuri added 10 commits September 3, 2024 21:10

add test

d19b595

Signed-off-by: Nikita Vemuri <[email protected]>

Merge branch 'master' of https://github.com/nikitavemuri/ray into nik…

9a16a2e

…ita/write_job_submission_events

update

4794342

Signed-off-by: Nikita Vemuri <[email protected]>

pass logdir

ce59bc9

Signed-off-by: Nikita Vemuri <[email protected]>

update

60e23aa

Signed-off-by: Nikita Vemuri <[email protected]>

update

bfefd83

Signed-off-by: Nikita Vemuri <[email protected]>

update build file rules

9fd8989

Signed-off-by: Nikita Vemuri <[email protected]>

merge

240b99c

Signed-off-by: Nikita Vemuri <[email protected]>

merge

aa280e8

Signed-off-by: Nikita Vemuri <[email protected]>

add comments

5e462f9

Signed-off-by: Nikita Vemuri <[email protected]>

nikitavemuri changed the title ~~[WIP] Write job submission events~~ [observability][export-api] Write submission job events Sep 5, 2024

nikitavemuri marked this pull request as ready for review September 5, 2024 20:05

nikitavemuri requested a review from rkooo567 September 5, 2024 20:05

nikitavemuri assigned rkooo567 Sep 5, 2024

nikitavemuri requested review from alanwguo and brycehuang30 September 5, 2024 22:16

alanwguo reviewed Sep 5, 2024

View reviewed changes

rkooo567 reviewed Sep 5, 2024

View reviewed changes

Nikita Vemuri added 5 commits September 5, 2024 17:49

update

a15a690

Signed-off-by: Nikita Vemuri <[email protected]>

merge

4d6394b

Signed-off-by: Nikita Vemuri <[email protected]>

update test

3de3653

Signed-off-by: Nikita Vemuri <[email protected]>

formatting

9dd473d

Signed-off-by: Nikita Vemuri <[email protected]>

formatting

0412688

Signed-off-by: Nikita Vemuri <[email protected]>

Nikita Vemuri added 6 commits September 19, 2024 13:13

formatting

5f41b82

Signed-off-by: Nikita Vemuri <[email protected]>

formatting

8d6ff9d

Signed-off-by: Nikita Vemuri <[email protected]>

update

4790c6c

Signed-off-by: Nikita Vemuri <[email protected]>

formatting

65df284

Signed-off-by: Nikita Vemuri <[email protected]>

update

3d91b92

Signed-off-by: Nikita Vemuri <[email protected]>

move test

3e725c9

Signed-off-by: Nikita Vemuri <[email protected]>

nikitavemuri requested review from alanwguo and rkooo567 September 20, 2024 17:01

alanwguo approved these changes Sep 23, 2024

View reviewed changes

rename log dir

d44d9ba

Signed-off-by: Nikita Vemuri <[email protected]>

rkooo567 enabled auto-merge (squash) September 23, 2024 23:41

rkooo567 merged commit d2982b7 into ray-project:master Sep 23, 2024
6 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[observability][export-api] Write submission job events #47468

[observability][export-api] Write submission job events #47468

nikitavemuri commented Sep 3, 2024 •

edited

Loading

alanwguo Sep 5, 2024

alanwguo Sep 5, 2024

rkooo567 Sep 5, 2024

nikitavemuri Sep 6, 2024

nikitavemuri Sep 20, 2024

alanwguo Sep 5, 2024

nikitavemuri Sep 20, 2024

rkooo567 Sep 5, 2024

nikitavemuri Sep 19, 2024

rkooo567 Sep 5, 2024

rkooo567 Sep 5, 2024

rkooo567 Sep 5, 2024

rkooo567 Sep 5, 2024

rkooo567 Sep 5, 2024

rkooo567 Sep 5, 2024

nikitavemuri Sep 20, 2024

alanwguo Sep 23, 2024

rkooo567 Sep 5, 2024

nikitavemuri Sep 20, 2024

rkooo567 Sep 5, 2024

nikitavemuri Sep 20, 2024

rkooo567 Sep 5, 2024

alanwguo Sep 23, 2024

		ExportEventDataType = Union[ExportSubmissionJobEventData]


		def get_event_id():

		# Force flush so that we won't lose events
		self.logger.handlers[0].flush()

		_export_event_logger = {}


		def get_export_event_logger(source: ExportEvent.SourceType, sink_dir: str):

	if type(event_data) is ExportSubmissionJobEventData:
	if isinstance(event_data, ExportSubmissionJobEventData):

		self._gcs_aio_client = gcs_aio_client
		self._export_submission_job_event_logger = None

[observability][export-api] Write submission job events #47468

[observability][export-api] Write submission job events #47468

Conversation

nikitavemuri commented Sep 3, 2024 • edited Loading

Why are these changes needed?

Related issue number

Checks

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nikitavemuri commented Sep 3, 2024 •

edited

Loading